package com.deep.common;

import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.SchemaBuilder;
import io.debezium.spi.converter.CustomConverter;
import io.debezium.spi.converter.RelationalColumn;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.util.Properties;

/**
 * 自定义MySQL datetime类型转换器
 */
public class CustomTimeConvert implements CustomConverter<SchemaBuilder, RelationalColumn> {
    private static final Logger log = LoggerFactory.getLogger(CustomTimeConvert.class);

    // 默认格式
    private DateTimeFormatter datetimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    // 默认时区，如果设置了serverTimeZone参数，则为serverTimeZone参数的值
    private ZoneId timezoneId = ZoneId.systemDefault();

    @Override
    public void configure(Properties properties) {
        String datetimeFormat = properties.getProperty("datetime.formatter");

        if (datetimeFormat != null && datetimeFormat.trim().length() > 0) {
            datetimeFormatter = DateTimeFormatter.ofPattern(datetimeFormat.trim());
        } else {
            log.warn("datetime.formatter is null, use default formatter: yyyy-MM-dd HH:mm:ss");
        }
    }

    @Override
    public void converterFor(RelationalColumn relationalColumn, ConverterRegistration<SchemaBuilder> converterRegistration) {
        // 获取字段类型
        String sqlType = relationalColumn.typeName().toUpperCase();

        SchemaBuilder schemaBuilder = null;
        Converter converter = null;

        // 对datetime类型进行自定义转换
        if ("DATETIME".equals(sqlType)) {
            schemaBuilder = SchemaBuilder.string().optional().name("debezium.mysql.datetime.string");
            converter = this::convertDatetime;
        }

        // 注册
        if (schemaBuilder != null) {
            converterRegistration.register(schemaBuilder, converter);
        }
    }

    // 格式化datetime，全量阶段以java.sql.Timestamp接收，增量阶段以Java.time.LocalDateTime接收
    private String convertDatetime(Object input) {
        if (input != null) {
            if (input instanceof Timestamp) {
                Timestamp timestamp = (Timestamp) input;
                LocalDateTime localDateTime = timestamp.toLocalDateTime();
                return datetimeFormatter.format(localDateTime);
            } else if (input instanceof LocalDateTime) {
                return datetimeFormatter.format((LocalDateTime) input);
            } else {
                log.warn("field is not java.sql.Timestamp or java.time.LocalDateTime, current field type is {}", input.getClass().toString());
                return null;
            }
        } else {
            return null;
        }
    }
}